package com.atguigu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.KafkaUtil_wm;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Dws_Trade_SourceWindow {
    public static void main(String[] args) {
        // TODO: 2023/3/28 需求：交易主题 1.4.1 各来源交易统计 来源粒度 指标：销售额，下单人数，下单次数，转化率 
        // TODO: 2023/3/28 获取flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism( 1 );
        // TODO: 2023/3/28 读取 dwd层 3表关联结果表 主题数据   dwd_trade_source_reday 封装为流
        String topic = "dwd_trade_source_reday";
        String groupId = "Dws_Trade_SourceWindow";
        DataStreamSource<String> kafkaDS = env.addSource( KafkaUtil_wm.getFlinkKafkaConsumer( topic, groupId ) );
        // TODO: 2023/3/29 转为 json 对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map( new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject( value );
                return jsonObject;
            }
        } );
        // TODO: 2023/3/29 转为 javabean对象

        // TODO: 2023/3/29 关联维表 base_source    id     source_site
        // TODO: 2023/3/29 开窗聚合
        // TODO: 2023/3/29 写入 clickhouse
        // TODO: 2023/3/29 启动任务

    }
}
